package msgcomm.transceivers;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.TreeSet;

import msgcomm.CrashedException;
import msgcomm.Hooks;
import msgcomm.Receiver;
import msgcomm.messages.Message;
import msgcomm.messages.MulticastMessage;
import msgcomm.messages.Payload;
import msgcomm.messages.UnicastMessage;

import common.Logger;

/**
 * Implementa il multicast affidabile.
 * 
 * TODO migliorare efficienza nell'uso della memoria! 
 */
public class ReliableMulticast extends Unicast {
	
	private MyBasicMulticast trans;
	private HashMap<String, TreeSet<MulticastMessage>> outOfOrder; // messaggi ricevuti ma fuori ordine
	private HashMap<String, Integer> received; // messaggi ricevuti in ordine: conservo solo mcSeqNum dell'ultimo
	
	// COSTRUTTORI
	
	protected ReliableMulticast(String name, Queue<Message> msgQueue, Hooks reqList) {
		super(name, msgQueue, reqList);
		trans= new MyBasicMulticast(name, msgQueue, reqList);
		trans.setLogger(log);
		outOfOrder= new HashMap<String, TreeSet<MulticastMessage>>();
		received= new HashMap<String, Integer>();
	}
	
	protected ReliableMulticast(String name, Queue<Message> msgQueue) {
		this(name, msgQueue, null);
	}
	
	public ReliableMulticast(String name) {
		this(name, new LinkedList<Message>(), null);
	}
	
	public ReliableMulticast(String name, Hooks reqList) {
		this(name, new LinkedList<Message>(), reqList);
	}
	

	@Override
	public void setLogger(Logger log) {
		super.setLogger(log);
		trans.setLogger(log);
	}
	
	// METODI PER IL MULTICAST
	
	public void rMulticast(Map<String, Receiver> dests, Payload data) throws CrashedException {
		MulticastMessage msg = new MulticastMessage(name,data,dests);
		trans.bMulticast(dests, msg);
	}
	
	protected void rMulticast(MulticastMessage msg) throws CrashedException {
		log.printLogTh("r-multicast " + msg);
		trans.bMulticast(msg.getDestinations(), msg);
	}
	
	protected void rDeliver(Message msg) {
		log.printLogTh("r-deliver " + msg);
		putMessage(msg);
	}
	
	@Override
	public void receive(Message msg) {
		log.printLogTh("r-receive " + msg);
		// scendo fino al transceiver base
		trans.receive(msg);
	}
	
	protected boolean yetDelivered(MulticastMessage msg) {
		String src= msg.getSender();
		int seq= msg.getMcSeqNum();
		boolean ret= false;
		
		if (received.containsKey(src)) {
			int val= received.get(src);
			if (seq <= val) {
				// gia' ricevuto (in sequenza)
				ret= true;
			}
		}
		if (!ret && outOfOrder.containsKey(src)) {
			TreeSet<MulticastMessage> set= outOfOrder.get(src);
			if (set.contains(msg)) {
				ret= true;
			}
		}
		return ret;
	}
	
	protected void addToDelivered(MulticastMessage msg) {
		String src= msg.getSender();
		int seq= msg.getMcSeqNum();
		
		if (received.containsKey(src)) {
			int val= received.get(src);
			if (val+1 == seq) {
				// ricevuto in sequenza
				received.put(src, seq);
				
				updateReceived(src);
				
			} else {
				//ricevuto fuori sequenza
				TreeSet<MulticastMessage> set= outOfOrder.get(src);
				set.add(msg);
				
				log.printLog("r-multicast: messaggio da " + src + " fuori sequenza. Sono presenti " + set.size() + " messaggi fuori sequenza da questo mittente");
				//log.printLog(received.toString());
				//log.printLog(outOfOrder.toString());
			}
		} else {
			// nuovo mittente
			log.printLog("r-multicast: nuovo mittente " + src);
			received.put(src, seq);
			TreeSet<MulticastMessage> set= new TreeSet<MulticastMessage>(); // set vuoto
			outOfOrder.put(src, set);
		}
	}
	
	protected void updateReceived(String src) {
		// verifico se posso togliere dei msg dall'insieme fuori sequenza
		TreeSet<MulticastMessage> set= outOfOrder.get(src);
		
		MulticastMessage msg;
		int seqNum= received.get(src);
		boolean changed= false;
		boolean cont= true;
		
		while (cont) {
			if (!set.isEmpty()) {
				msg= set.first();
				if (msg.getMcSeqNum() <= seqNum) {
					// vecchio msg
					set.remove(msg);
					changed= true;
				} else if (seqNum+1 == msg.getMcSeqNum()) {
					// msg in sequenza
					seqNum++;
					set.remove(msg);
					changed= true;
				} else {
					cont= false; // il primo msg non e' in sequenza
				}
			} else {
				cont= false; // non ci sono piu' msg nel set
			}
		}
		if (changed) {
			// aggiorno
			received.put(src, seqNum);
			log.printLog("r-multicast: sono rimasti " + set.size() + " messaggi fuori sequenza provenienti da " + src);
		}
	}
	
	public synchronized void clearReceivedSet() {
		received.clear();
		outOfOrder.clear();
		log.printLog("r-multicast: clear");
	}
	
	
	/**
	 * Rimuove un determinato transceiver dalle mappe dei messaggi ricevuti in ordine e di quelli fuori ordine
	 * @param transName il nome del transceiver da rimuovere dalle mappe
	 */
	public synchronized void removeTransReceivedSet(String transName){
		
		received.remove(transName);
		outOfOrder.remove(transName);
		
	}
	
	
	public synchronized void forceReceivedSet(String name, int mcSeqNum) {
		received.put(name, mcSeqNum);
		if (!outOfOrder.containsKey(name)) {
			outOfOrder.put(name, new TreeSet<MulticastMessage>());
		}
		log.printLog("r-multicast: forza " + name + " a " + mcSeqNum + ": " + received.toString());
		updateReceived(name);
	}

	
	private class MyBasicMulticast extends BasicMulticast {

		protected MyBasicMulticast(String name, Queue<Message> msgQueue, Hooks reqList) {
			super(name, msgQueue, reqList);
		}
		
		@Override
		protected void bDeliver(Message msg) {
			// risalgo la catena di deliver
			log.printLogTh("b-deliver(override) " + msg);
			if (msg instanceof UnicastMessage) {
				super.uniDeliver(msg);
			} else if (msg instanceof MulticastMessage) {
				synchronized (ReliableMulticast.this) {
					onBDeliver((MulticastMessage)msg);
				}
			} else {
				throw new RuntimeException("Posso ricevere solo messaggi unicast e multicast");
			}
		}
		
		private void onBDeliver(MulticastMessage msg) {
			if (!yetDelivered(msg)) {
				addToDelivered(msg);
				if (!msg.getSender().equals(name)) {
					// io non sono il mittente
					log.printLogTh("b-deliver(Override): lancio propagazione bMulticast in background di " + msg);
					final MulticastMessage fMsg= msg;
					Thread t= new Thread(new Runnable() {
						@Override
						public void run() {
							log.printLogTh("b-deliver(Override): esecuzione propagazione bMulticast di " + fMsg);
							//rMulticast(msg); // uso rMulticast() per esegue il controllo sulla mappa dei destinatari
							try {
								Map<String, Receiver> dests= new LinkedHashMap<String, Receiver>(fMsg.getDestinations());
								dests.remove(name); // non propago verso me stesso
								dests.remove(fMsg.getSender()); // non propago verso il sender
								bMulticast(dests, fMsg);
							} catch (Exception e) {
								// ignoro!
								// TODO sicuri?
							}
						}
					});
					// nessun problema in esecuzioni concorrenti della propagazione
					t.setPriority(Thread.MIN_PRIORITY);
					t.start();
				}
				rDeliver(msg);
			} else {
				log.printLogTh("b-deliver(Override) " + msg + " gia' ricevuto");
			}
		}
	}
}
